/**
 * @开发版权 
 * @项目名称 轻量数据中台(LDMP)
 * @版本信息 v1.0
 * @开发人员 zhous
 * @开发日期 2025-01-21
 * @修订日期
 * @描述  Application
 */
package com.ng.common.component.mqtt;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.ng.biz.service.IMqttClient;
import com.ng.biz.service.impl.MqttClientImpl;
import com.ng.common.component.mqtt.callback.LDMPMqttCallback;
import com.ng.common.util.SpringContextHolder;
import com.ng.common.util.UuidUtil;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class LDMPMqttClient {
	
	private IMqttClient mqttClientImpl;
	
    private MqttClient client;
    private String taskId;
    private String host;
    private String username;
    private String password;
    private String clientPrefix;
    private String clientId;
    private int timeout;
    private int keepalive;
    private com.ng.biz.model.domain.MqttClient mqttClient;
    
    /**
     * mqtt客户端类型:0-消费者 1-生产者
     * 考虑对于 消费者是否记录消费日志
     * 对于生产者是否记录消息是否送达日志
     */
    private String type;
    
	private final Set<String> subscribedTopics = Collections.synchronizedSet(new HashSet<>()); // 线程安全的已订阅主题集合
	
	private final ConcurrentMap<String, JSONObject> topicMapingFlow = new ConcurrentHashMap<>();

    private final long RECONNECT_INTERVAL_MS = 5000; // 重连间隔时间，单位为毫秒

    private Thread reconnectThread; // 用于重连的守护线程

    private volatile boolean reconnecting = false; // 用于控制重连操作的标志位

    public LDMPMqttClient() {
		clientId=clientPrefix+"_"+UuidUtil.genCode();
		mqttClientImpl=SpringContextHolder.getBean(MqttClientImpl.class);
	}
    
    public LDMPMqttClient(String taskId) {
		clientId=clientPrefix+"_"+UuidUtil.genCode();
		mqttClientImpl=SpringContextHolder.getBean(MqttClientImpl.class);
	}
    
    public LDMPMqttClient(String taskId,String host, String username, String password, String clientPrefix, com.ng.biz.model.domain.MqttClient mqttClient, int timeout, int keepalive) {
		super();
		mqttClientImpl=SpringContextHolder.getBean(MqttClientImpl.class);
		this.taskId = taskId;
		this.host = host;
		this.username = username;
		this.password = password;
		this.clientPrefix = clientPrefix;
		this.clientId = clientPrefix+"_"+UuidUtil.genCode();;
		this.mqttClient = mqttClient;
		this.timeout = timeout;
		this.keepalive = keepalive;
		this.type = mqttClient.getType();
	}

    public MqttClient getClient() {
        return client;
    }
    
    public String getClientId() {
		return clientId;
	}

	public String getType() {
		return type;
	}
	

	public String getTaskId() {
		return taskId;
	}

	public ConcurrentMap<String, JSONObject> getTopicMapingTopic() {
		return topicMapingFlow;
	}

	/**
     * 设置mqtt连接参数
     * @param username
     * @param password
     * @param timeout
     * @param keepalive
     * @return
     */
    public MqttConnectOptions setMqttConnectOptions(String username, String password, int timeout, int keepalive) {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(timeout);
        options.setKeepAliveInterval(keepalive);
        options.setCleanSession(true);
        options.setAutomaticReconnect(true);
        options.setMaxInflight(2000);
        return options;
    }
    
    
    /**
     * 连接mqtt服务端，得到MqttClient连接对象
     * @return
     * @throws MqttException
     */
    public boolean connect() throws MqttException {
    	
        if (client == null) {
            client = new MqttClient(host,clientId, new MemoryPersistence());
            client.setCallback(new LDMPMqttCallback(this));
        }
        MqttConnectOptions mqttConnectOptions = setMqttConnectOptions(username, password, timeout, keepalive);
        if (!client.isConnected()) {
            client.connect(mqttConnectOptions);
        } else {
            client.disconnect();
            client.connect(mqttConnectOptions);
        }
        
    	
    	if(mqttClient!=null) {
    		JSONArray topics=JSONArray.parseArray(mqttClient.getTopics());
    		if(topics!=null && topics.size()>0) {
    			for(Object obj:topics) {
    				JSONObject topic=(JSONObject)obj;
    				if(topic!=null) {
    					topicMapingFlow.put(topic.getString("prefix"), topic);
    					subscribe(topic.getString("topic"), topic.getIntValue("qos"));
    				}
    			}
    		}
    	}
    	log.info("{} connect success.",clientId);
		return true;
    }
    
    public void subscribeTopics() {
		synchronized (subscribedTopics) {
			for (String topic : subscribedTopics) {
				try {
					log.info("{} : 重连成重新订阅主题 : {}",clientId,topic);
					subscribe(topic);
				} catch (MqttException e) {
					log.error("{} : 重连成重新订阅主题异常 : {}",clientId,e.getMessage());
				} 
			}
		}
    }
    
	public void subscribe(String topic) throws MqttException {
		client.subscribe(topic);
		synchronized (subscribedTopics) {
			subscribedTopics.add(topic); 
		}
		log.info("Re-subscribed to topic: {} for client: {}", topic, clientId);
	}
	
	public void subscribe(String topic, int qos ) throws MqttException {
		client.subscribe(topic,qos);
		synchronized (subscribedTopics) {
			subscribedTopics.add(topic); 
		}
		log.info("Re-subscribed to topic: {} for client: {}", topic, clientId);
	}
	
    /**
     * 发布消息
     *
     * @param pushMessage
     * @param topic
     * @param qos
     * @param retained:留存
     */
    public void publish(String topic, String payload, int qos, boolean retained) {
		CompletableFuture.runAsync(() -> {
			try {
				MqttMessage message = new MqttMessage(payload.getBytes());
				message.setQos(qos); 
				message.setRetained(retained);
				this.client.publish(topic, message);
				mqttClientImpl.sendIncrementAndGet();
			} catch (MqttException e) {
				log.error(" 信息推送异常: client: {} - Topic:{} - error:{}",clientId,topic,e.getMessage());
			}
		});
    }
	
    /**
     * 取消订阅主题
     * @param topic 要取消订阅的主题
     * @throws MqttException
     */
    public void unsubscribe(String topic) throws MqttException {
        if (client != null && client.isConnected()) {
            client.unsubscribe(topic);
            synchronized (subscribedTopics) {
                subscribedTopics.remove(topic);
            }
            log.info("Unsubscribed from topic: {} for client: {}", topic, clientId);
        } else {
            log.warn("Cannot unsubscribe from topic: {} because client is not connected.", topic);
        }
    }

    /**
     * 启动MQTT重连守护进程
     */
    public void startReconnectDaemon() {
        reconnectThread = new Thread(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    if (!reconnecting && (client == null || !client.isConnected())) {
                        reconnecting = true;
                        log.info("MQTT client is not connected. Attempting to reconnect...");
                        connect();
                        subscribeTopics(); // 重新订阅主题
                        reconnecting = false;
                    }
                    Thread.sleep(RECONNECT_INTERVAL_MS);
                } catch (InterruptedException e) {
                    log.error("Reconnect thread interrupted: {}", e.getMessage());
                    Thread.currentThread().interrupt();
                } catch (MqttException e) {
                    log.error("Error while reconnecting MQTT client: {}", e.getMessage());
                    reconnecting = false;
                }
            }
        });

        reconnectThread.setDaemon(true); // 设置为守护线程
        reconnectThread.start();
    }

    /**
     * 断开MQTT连接并停止守护进程
     */
    public void disconnect() {
        try {
            if (client != null && client.isConnected()) {
                client.disconnect();
                log.info("MQTT client disconnected: {}", clientId);
            }
        } catch (MqttException e) {
            log.error("Error while disconnecting MQTT client: {}", e.getMessage());
        }

        if (reconnectThread != null && reconnectThread.isAlive()) {
            reconnectThread.interrupt(); // 中断守护线程
            log.info("Reconnect daemon thread stopped.");
        }
    }

    /**
     * 检查是否正在重连
     * @return true if reconnecting, false otherwise
     */
    public boolean isReconnecting() {
        return reconnecting;
    }

    /**
     * 设置重连状态
     * @param reconnecting the reconnecting status to set
     */
    public void setReconnecting(boolean reconnecting) {
        this.reconnecting = reconnecting;
    }
}
